#include <iostream>
#include "Sunnet.h"
#include <assert.h>

#include <unistd.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>

using namespace std;

Sunnet* Sunnet::inst;
Sunnet::Sunnet(){
    inst = this;
}

//开启系统
void Sunnet::Start(){
    cout << "[Sunnet][Start]" << endl;
    //忽略SIGPIPE信号
    signal(SIGPIPE, SIG_IGN);
    //锁
    pthread_rwlock_init(&servicesLock, NULL);
    pthread_spin_init(&globalLock, PTHREAD_PROCESS_PRIVATE);

    pthread_mutex_init(&sleepMtx, NULL);
    pthread_cond_init(&sleepCond, NULL);

    //开启Worker
    StartWorker();
    //开启Socket线程
    StartSocket();
    assert(pthread_rwlock_init(&connsLock, NULL)==0);
}

void Sunnet::Wait(){
    if(workerThreads[0]){
        workerThreads[0]->join();
    }
}

//创建服务
uint32_t Sunnet::NewService(shared_ptr<string> type){
    auto srv = make_shared<Service>();
    srv->type = type;
    pthread_rwlock_wrlock(&servicesLock);
    {
        srv->id = maxId;
        maxId++;
        services.emplace(srv->id, srv);
    }
    pthread_rwlock_unlock(&servicesLock);
    srv->OnInit();
    return srv->id;
}

//获取服务
shared_ptr<Service> Sunnet::GetService(uint32_t id){
    shared_ptr<Service> srv = NULL;
    pthread_rwlock_rdlock(&servicesLock);
    {
        unordered_map<uint32_t, shared_ptr<Service>>::iterator iter = services.find(id);
        if(iter != services.end()){
            srv = iter->second;
        }
    }
    pthread_rwlock_unlock(&servicesLock);
    return srv;
}

//删除服务
//只能service自己调自己,因为会调用不加锁的srv->OnExit和srv->isExiting
void Sunnet::KillService(uint32_t id){
    cout << "[Sunnet][KillService] id:" << id << endl;
    shared_ptr<Service> srv = GetService(id);
    if(!srv){
        cout << "[Sunnet][KillService] fail" << endl;
        return;
    }
    //退出前
    srv->OnExit();
    srv->isExiting = true;
    //删除列表
    pthread_rwlock_wrlock(&servicesLock);
    {
        services.erase(id);
    }
    pthread_rwlock_unlock(&servicesLock);
}

int Sunnet::Listen(uint32_t port, uint32_t serviceId){
    cout << "[Sunnet][Listen] service:" << serviceId << " port:" << port << endl;
    //步骤1:创建socket
    int listenFd = socket(AF_INET, SOCK_STREAM, 0);
    if(listenFd <= 0){
        cout << "[Sunnet][Listen] error, listenFd <= 0 fd=" << listenFd << endl;
        return -1;
    }
    //步骤2:设置为非阻塞
    fcntl(listenFd, F_SETFL, O_NONBLOCK);
    //步骤3:设置端口复用
    int optval = 1;
    if(setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval))){
        cout << "[Sunnet][Listen] error, setsockopt SO_REUSEADDR failed" << endl;
        close(listenFd);
        return -1;
    }
    //步骤3:bind
    struct sockaddr_in addr; //创建地址结构
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    int r = bind(listenFd, (struct sockaddr *)&addr, sizeof(addr));
    if(r == -1){
        close(listenFd);
        cout << "[Sunnet][Listen] error bind fail r=" << r << endl;
        return -1;
    }
    //步骤4:listen
    r = listen(listenFd, 64);
    if(r < 0){
        close(listenFd);
        cout << "[Sunnet][Listen] fail r=" << r << endl;
        return -1;
    }
    //步骤5:添加到管理结构
    AddConn(listenFd, serviceId, Conn::TYPE::LISTEN);
    //步骤6:epoll事件,跨线程
    socketWorker->AddEvent(listenFd);
    return listenFd;
}


void Sunnet::StartWorker(){
    for(int i=0; i<WORKER_NUM; i++){
        cout << "[Sunnet][StartWorker] worker thread:" << i << endl;
        Worker* worker = new Worker();
        worker->id = i;
        worker->eachNum = 2 << i;

        thread *wt = new thread(*worker);
        workers.push_back(worker);
        workerThreads.push_back(wt);
    }
}

//开启Socket线程
void Sunnet::StartSocket()
{
    //创建线程对象
    socketWorker = new SocketWorker();
    //初始化
    socketWorker->Init();
    //创建线程
    socketThread = new thread(*socketWorker);
}

//检查并唤醒线程
void Sunnet::CheckAndWeakUp(){
    //没有休眠的工作线程
    //这里可能存在多线程同时访问sleepCount,由于没加锁sleepCount的值可能不准确，
    //尽管出现的概率较小，但不排除某次需要唤醒时没有唤醒(因为sleepCount的值小了)
    //或不该唤醒时唤醒了(sleepCount的值大了).但这两种错误无伤大雅,若出现唤醒时没有唤醒的情况,
    //后续还会加上Socket模块,接收数据时也要唤醒工作线程.
    //等待下一次唤醒即可(不该唤醒时唤醒)更没关系,就当工作线程空置一次.
    if(sleepCount == 0){
        return;
    }
    //WORKER_NUM-sleepCount=当前正在工作的线程
    //正在工作的线程是否足够。比如系统中只有2个待处理的服务(globalLen),
    //而系统开启了5条工作线程(WORKER_NUM),
    //目前只有1条线程在休眠(sleepCount),
    //那剩下的4条线程正在工作,足以应对.
    if(WORKER_NUM - sleepCount <= globalLen){
        pthread_cond_signal(&sleepCond);
    }
}

//Worker线程调用,进入休眠
void Sunnet::WorkerWait(){
    pthread_mutex_lock(&sleepMtx);
    //休眠数量++
    sleepCount++;
    //等待唤醒(pthread_cond_signal解锁唤醒)
    pthread_cond_wait(&sleepCond, &sleepMtx);
    //休眠数量--
    sleepCount--;
    pthread_mutex_unlock(&sleepMtx);
}

void Sunnet::Send(uint32_t told, shared_ptr<BaseMsg> msg){
    shared_ptr<Service> toSrv = GetService(told);
    if(!toSrv){
        return;
    }
    //插入目标服务的消息队列
    toSrv->PushMsg(msg);
    //检查并放入全局队列
    bool hasPush = false;
    pthread_spin_lock(&toSrv->inGlobalLock);
    {
        if(!toSrv->inGlobal){
            PushGlobalQueue(toSrv);
            toSrv->inGlobal = true;
            hasPush = true;
        }
    }
    pthread_spin_unlock(&toSrv->inGlobalLock);
    //唤起进程
    if(hasPush){
        CheckAndWeakUp();
    }
}

//弹出全局队列
shared_ptr<Service> Sunnet::PopGlobalQueue(){
    shared_ptr<Service> srv = NULL;
    pthread_spin_lock(&globalLock);
    {
        if(!globalQueue.empty()){
            srv = globalQueue.front();
            globalQueue.pop();
            globalLen--;
        }
    }
    pthread_spin_unlock(&globalLock);
    return srv;
}

void Sunnet::PushGlobalQueue(shared_ptr<Service> srv){
    pthread_spin_lock(&globalLock);
    {
        globalQueue.push(srv);
        globalLen++;
    }
    pthread_spin_unlock(&globalLock);
}

shared_ptr<BaseMsg> Sunnet::MakeMsg(uint32_t source, char *buff, int len){
    auto msg = make_shared<ServiceMsg>();
    msg->type = BaseMsg::TYPE::SERVICE;
    msg->source = source;
    //基本类型的对象没有析构函数
    //所以用delete或delete[]都可以销毁基本类型数组
    //智能指针默认使用delete销毁对象
    //所以无须重写智能指针的铛方法
    msg->buff = shared_ptr<char>(buff);
    msg->size = len;
    return msg;
}

//创建连接
int Sunnet::AddConn(int fd, uint32_t id, Conn::TYPE type){
    auto conn = make_shared<Conn>();
    conn->fd = fd;
    conn->serviceId = id;
    conn->type = type;
    pthread_rwlock_wrlock(&connsLock);
    {
        conns.emplace(fd, conn);
    }
    pthread_rwlock_unlock(&connsLock);
    return fd;
}

//获取连接
shared_ptr<Conn> Sunnet::GetConn(int fd){
    shared_ptr<Conn> conn = NULL;
    pthread_rwlock_rdlock(&connsLock);
    {
        unordered_map<uint32_t, shared_ptr<Conn>>::iterator iter = conns.find(fd);
        if(iter != conns.end()){
            conn = iter->second;
        }
    }
    pthread_rwlock_unlock(&connsLock);
    return conn;
}

//删除连接
bool Sunnet::RemoveConn(int fd)
{
    int result;
    pthread_rwlock_wrlock(&connsLock);
    {
        result = conns.erase(fd);
    }
    pthread_rwlock_unlock(&connsLock);
    return result = 1;
}

//关闭连接
void Sunnet::CloseConn(uint32_t fd){
    cout << "[Sunnet][CloseConn] fd:" << fd << endl;
    //删除对象
    bool succ = RemoveConn(fd);
    //关闭套接字
    close(fd);
    //删除epoll对象对套接字的监听(跨线程)
    if(succ){
        socketWorker->RemoveEvent(fd);
    }
}

void Sunnet::ModifyEvent(int fd, bool epollOut) {
    socketWorker->ModifyEvent(fd, epollOut);
}